package com.cxs.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;

/**
 * @Author: cxs
 */
@Configuration
@Slf4j
public class MyKfkConsumerConfig {


    /**
     * consumer消费失败后的错误处理
     * kafka，自身不支持重复消费，需要自己写重试机制
     *
     * @return
     */
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            System.out.println("消费异常：" + message.getPayload());
            // 这里可以吧消息签收了，不然消息依然在分区中 等待下次启动再消费
            consumer.commitSync();
            return null;
        };
    }

    /**
     * 消息过滤
     * 根据业务来控制
     *
     * @return
     */
    @Bean
    public RecordFilterStrategy<?, ?> recordFilterStrategy() {
        return new RecordFilterStrategy<Object, Object>() {
            @Override
            public boolean filter(ConsumerRecord<Object, Object> consumerRecord) {
                String s = consumerRecord.value().toString();
                if (s.equals("第1个消息")) {
                    return false;
                }
                // 奇葩 返回true 表示需要过滤掉 也就是说不会再消费者看到。只有返回false 才会被消费者消费
                return true;
            }
        };
    }

}
